package com.atguigu.flink.sql;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Created by Smexy on 2022/12/21
 */
public class Demo11_TVF
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.getConfig().setAutoWatermarkInterval(2000);

        env.setParallelism(1);

        //为了获取eventtime
        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy.<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, r) -> e.getTs());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            .assignTimestampsAndWatermarks(watermarkStrategy)
            ;


        Table table = tableEnv.fromDataStream(ds, $("ts"),$("vc"),$("id"),$("pt").proctime(),
            $("et").rowtime());

        //给表起名字
        tableEnv.createTemporaryView("ws",table);

        //滚动
        String tumbleSql = "SELECT window_start, window_end, id , SUM(vc) sumVC" +
            "  FROM TABLE(" +
            "    TUMBLE(TABLE ws, DESCRIPTOR(et), INTERVAL '5' SECONDS))" +
            "  GROUP BY window_start, window_end, id";

        //滑动
        String slideSql = "SELECT window_start, window_end, id , SUM(vc) sumVC" +
            "  FROM TABLE(" +
            "    HOP(TABLE ws, DESCRIPTOR(et), INTERVAL '5' SECONDS , INTERVAL '10' SECONDS))" +
            "  GROUP BY window_start, window_end, id";

        //1.13不支持session窗口。
        /*
                累积窗口。
                    每隔1h统计过去1天中累积PV
                     1:   0-1 PV
                     2:   0-1 + 1-2 PV
                     3:   0-1 + 1-2 + 2-3 PV

                     第一个 INTERVAL 是滑动步长     1h
                     第二个 INTERVAL 是窗口的最大范围  24h
         */
        String culSql = "SELECT window_start, window_end, id , SUM(vc) sumVC" +
            "  FROM TABLE(" +
            "    CUMULATE(TABLE ws, DESCRIPTOR(et), INTERVAL '2' SECONDS , INTERVAL '6' SECONDS))" +
            "  GROUP BY window_start, window_end, id";

        tableEnv.executeSql(culSql)
                .print();

    }
}
